package com.atcyj.gmall.realtime.dim.app;

import com.alibaba.fastjson.JSONObject;
import com.atcyj.gamll.realtime.common.base.BaseApp;
import com.atcyj.gamll.realtime.common.bean.TableProcessDim;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.FlinkSourceUtil;
import com.atcyj.gamll.realtime.common.util.HBaseUtil;
import com.atcyj.gmall.realtime.dim.function.DimBroadcastFunction;
import com.atcyj.gmall.realtime.dim.function.HBaseSinkFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @author: cyj
 * @date: 2024/8/14
 */
@Slf4j
public class DimApp extends BaseApp {
    // @Slf4j
    //private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(DimApp.class);
    public static void main(String[] args) {
        new DimApp().start(10001, 4, "dim_app", Constant.TOPIC_DB);
        // 在 start()方法中调用子类实现的 handle()方法
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        // TODO 1. 对ods层原始数据进行数据清洗:过滤脏数据+转换数据类型

        SingleOutputStreamOperator<JSONObject> jsonObjStream = etl(stream);
        // TODO 2. 使用flinkCDC读取配置表数据

        MySqlSource<String> mySqlSource = FlinkSourceUtil.getMySqlSource(Constant.MYSQL_PROCESS_DATABASE, Constant.MYSQL_PROCESS_TABLE);
        DataStreamSource<String> mysqlSource = env
                .fromSource(
                        mySqlSource,
                        WatermarkStrategy.noWatermarks(),
                        "flink_cdc")
                .setParallelism(1); // mysql 只有一台服务器，并行度设置为1即可

        // TODO 3. 根据配置表数据在HBase中创建维度表
        SingleOutputStreamOperator<TableProcessDim> tableProcessDimStream = createHBaseTable(mysqlSource);


        // TODO 4. 将配置表数据流做成广播流
        //  MapStateDescriptor<UK, UV>
        MapStateDescriptor<String, TableProcessDim> broadcastState =
                new MapStateDescriptor<String, TableProcessDim>("broadcast_state", String.class, TableProcessDim.class);

        BroadcastStream<TableProcessDim> dimBroadcastStream = tableProcessDimStream.broadcast(broadcastState);

        // TODO 5. 连接主流和广播流
        BroadcastConnectedStream<JSONObject, TableProcessDim> connectedStream = jsonObjStream.connect(dimBroadcastStream);

        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimStream = connectedStream.process(
                new DimBroadcastFunction(broadcastState)).setParallelism(1);

        // TODO 6. 筛选出需要写出的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filteredColumnStream = fileterColumn(dimStream);


        // TODO 7. 写出到HBase
        filteredColumnStream.addSink(new HBaseSinkFunction());

    }

    public SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> fileterColumn(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimStream) {
        return dimStream.map(new MapFunction<Tuple2<JSONObject, TableProcessDim>, Tuple2<JSONObject, TableProcessDim>>() {
            @Override
            public Tuple2<JSONObject, TableProcessDim> map(Tuple2<JSONObject, TableProcessDim> value) throws Exception {

                JSONObject jsonObject = value.f0;
                TableProcessDim tableProcessDim = value.f1;

                List<String> columns = Arrays.asList(tableProcessDim.getSinkColumns().split(","));
                JSONObject data = jsonObject.getJSONObject("data");
                data.keySet().removeIf(key -> !columns.contains(key));

                return value;
            }
        });
    }

    public SingleOutputStreamOperator<TableProcessDim> createHBaseTable(DataStreamSource<String> mysqlSource) {
        SingleOutputStreamOperator<TableProcessDim> dimStream = mysqlSource.flatMap(new RichFlatMapFunction<String, TableProcessDim>() {
            Connection connection = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 获取HBase连接
                connection = HBaseUtil.getConnection();
            }

            @Override
            public void close() throws Exception {
                // 关闭连接
                HBaseUtil.closeConnection(connection);
            }

            @Override
            public void flatMap(String value, Collector<TableProcessDim> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);

                TableProcessDim dim;

                String op = jsonObject.getString("op");
                if ("c".equals(op) || "r".equals(op)) {
                    dim = jsonObject.getObject("after", TableProcessDim.class);
                    createTable(dim);
                } else if ("u".equals(op)) {
                    // 先删除 如果改了表名的话，由A改成B，那么应该把A和B都删掉，再重新建B表
                    // 删除after中的表
                    dim = jsonObject.getObject("after", TableProcessDim.class);
                    dropTable(dim);
                    // 重建after中的表
                    createTable(dim);
                    dim = jsonObject.getObject("before", TableProcessDim.class);
                    // 删除被修改的信息 即删除before中的表
                    dropTable(dim);
                } else {
                    dim = jsonObject.getObject("before", TableProcessDim.class);
                    dropTable(dim);
                }
                dim.setOp(op);
                out.collect(dim);
            }

            public void createTable(TableProcessDim dim) throws IOException {
                String[] families = dim.getSinkFamily().split(",");
                HBaseUtil.createTable(connection, Constant.HBASE_NAMESPACE, dim.getSinkTable(), families);
            }

            private void dropTable(TableProcessDim dim) throws IOException {
                HBaseUtil.dtopTable(connection, Constant.HBASE_NAMESPACE, dim.getSinkTable());
            }

        });
        return dimStream;
    }

    public SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String database = jsonObject.getString("database");
                    String type = jsonObject.getString("type");
                    JSONObject data = jsonObject.getJSONObject("data");
                    if ("gmall".equals(database)
                            && !"bootstrap-start".equals(type) && !"bootstrap-complete".equals(type)
                            && data != null && data.size() != 0) {
                        out.collect(jsonObject);
                    }

                } catch (Exception e) {
                    e.printStackTrace();

                }
            }
        });
    }
}
